package czk.concrent.threadpool.ownthreadpool;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 自定义线程池的实现
 * @Author:ChenZhangKun
 * @Date: 2020/6/23 21:30
 */
public class OwnerThreadPool {
    public static void main(String[] args) {
        ThreadPool threadPool=new ThreadPool(3,1000,TimeUnit.MILLISECONDS,10,((blockQueue, task) -> {
            // 1 死等
           // blockQueue.put(task);
            // 2 带超时等待
           // blockQueue.offer(task,500,TimeUnit.MILLISECONDS);
            // 3 放弃执行
           // System.out.println("放弃执行");
            // 4 调用者着抛出异常
            throw new RuntimeException("任务过多，放弃执行");
            // 5 调用者自己执行任务
            //task.run();
        }));
        for (int i = 0; i <10; i++) {
            int j=i;
            threadPool.execute(()->{
                System.out.println("执行任务  "+j);
            });
        }
    }

}
class ThreadPool{
    // 阻塞队列
    private final BlockQueue<Runnable> taskQueue;
    // 执行的线程集合
    private final HashSet<Worker> workers=new HashSet();
    // 核心数
    private final int coreNum;
    // 超时时间
    private final long timeout;
    // 时间单位
    private final TimeUnit timeUnit;
    // 构造方法中指定策略
    private final RejectPolicy<Runnable> rejectPolicy;
    public void execute(Runnable task){
        // 任务数没有超过核心coreNum时，直接交给worker执行
       synchronized (workers){
           if (workers.size()<coreNum){
               System.out.println("新增worker");
               // 交给worker处理
               Worker worker=new Worker(task);
               // 加入线程集合
               System.out.println("加入线程集合");
               this.workers.add(worker);
               // 执行线程方法
               worker.start();
           }else {
               //
               System.out.println("新增任务队列");
               // ①死等
               taskQueue.put(task);
               // ②超时时间等待
               // ③放弃该线程
               // ④抛出异常
               // ⑤调用者自己执行任务
               taskQueue.tryPut(rejectPolicy,task);
           }
       }
    }
    public ThreadPool(int coreNum, long timeout, TimeUnit timeUnit,int capcity,RejectPolicy<Runnable> rejectPolicy) {
        this.coreNum = coreNum;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.taskQueue=new BlockQueue<>(capcity);
        this.rejectPolicy=rejectPolicy;
    }

    class Worker extends Thread{
        // 任务对象
        private Runnable task;

        public Worker(Runnable task) {
            this.task = task;
        }

        @Override
        public void run() {
            // task不为空时，执行任务
            // task为空时，执行任务队列的任务
            // task为空时，再去阻塞队列取任务赋予task继续判断
            //while (task!=null||(task=taskQueue.take())!=null){
            while (task!=null||(task=taskQueue.poll(timeout,timeUnit))!=null){
                try {
                    System.out.println("正在执行"+task);
                    task.run();
                }catch (Exception ex){
                    ex.printStackTrace();
                }finally {
                    task=null;
                }
            }
            synchronized (workers){
                System.out.println("移除"+this);
                workers.remove(this);
            }
        }
    }
}


class BlockQueue<T>{
    // 双向链表
    private Deque<T> deque=new ArrayDeque<>();
    // 锁
    private ReentrantLock reentrantLock=new ReentrantLock();
    // 生产者条件变量
    private Condition fullWaitSet=reentrantLock.newCondition();
    // 消费者条件变量
    private Condition emptyWaitSet=reentrantLock.newCondition();
    // 队列容量
    private int capcity;
    // 获取元素
    public T take(){
        try {
            //多线程。并发
            reentrantLock.lock();
            // 元素为空时，去等待
            try {
                while (deque.isEmpty()){
                    // 进入休息室等待
                    emptyWaitSet.await();
                }
            }catch (Exception ex){
                ex.printStackTrace();
            }
            // 被唤醒后取出元素
            T t = deque.removeFirst();
            // 唤醒生产者队列
            fullWaitSet.signal();
            return t;
        } finally {
            // 解锁
            reentrantLock.unlock();
        }
    }
    // 添加进入队列
    public void put(T t){
        reentrantLock.lock();
        try {
            while (deque.size()==capcity){
                // 生产者进入休息室等待
                fullWaitSet.await();
            }
            // 唤醒后放进队列
            deque.addLast(t);
            // 唤醒等待的队列
            emptyWaitSet.signal();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            reentrantLock.unlock();
        }
    }
    // 带超时时间的阻塞添加
    public boolean offer(T t,long timeout,TimeUnit timeUnit){

        reentrantLock.lock();
        try {
            // 将时间转换成纳秒
            long nanos = timeUnit.toNanos(timeout);

            // 进入阻塞
            while (deque.size()==capcity){
                System.out.println("进入阻塞等待");
                try {
                    // 等待时间大于timeout
                    if (nanos<=0){
                        return false;
                    }
                    nanos=fullWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    return false;
                }
            }
            // 时间范围内，加入任务队列
            System.out.println("加入阻塞队列");
            this.put(t);
            emptyWaitSet.signal();
            return true;
        }finally {
            reentrantLock.unlock();
        }
    }

    // 队列大小
    public int size(){
        reentrantLock.lock();
        try {
            return deque.size();
        }finally {
            reentrantLock.unlock();
        }

    }
    // 带超时的阻塞获取
    public T poll(long time, TimeUnit timeUnit){
        try {
            // 将超时时间转为纳秒
            long nanos = timeUnit.toNanos(time);
            //多线程。并发
            reentrantLock.lock();
            // 元素为空时，去等待
            try {
                while (deque.isEmpty()){
                    // 进入休息室等待
                    if (nanos<=0){
                        return null;
                    }
                    // 返回需要等待的剩余时间
                    nanos=emptyWaitSet.awaitNanos(nanos);
                }
            }catch (Exception ex){
                ex.printStackTrace();
            }
            // 被唤醒后取出元素
            T t = deque.removeFirst();
            // 唤醒生产者队列
            fullWaitSet.signal();
            return t;
        } finally {
            // 解锁
            reentrantLock.unlock();
        }
    }

    public BlockQueue(int capcity) {
        this.capcity = capcity;
    }

    /**
     *
     * @param rejectPolicy
     * @param task
     */
    public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
        reentrantLock.lock();
        try {
            // 判断队列是都已满
            if (deque.size()==capcity){
                // 执行策略
                rejectPolicy.reject(this,task);
            }else {
                // 直接加入队列
                System.out.println("加入队列");
                deque.addLast(task);
                emptyWaitSet.signal();
            }
        }finally {
            reentrantLock.unlock();
        }
    }
}

/**
 * 策略模式，将超时等待选择的策略下放
 * 声明为函数式接口，配合lambda表达式的使用
 * @param <T>
 */
@FunctionalInterface
interface RejectPolicy<T>{
    void reject(BlockQueue<T> blockQueue,T task);
}
